package com.afdor.rws.event.driven.channel.consumer;

import com.afdor.rws.event.driven.EventHolder;
import com.afdor.rws.event.driven.Serialization;
import com.afdor.rws.event.driven.channel.consumer.handler.EventHandler;
import com.afdor.rws.event.driven.channel.consumer.handler.EventRetryStrategy;
import com.afdor.rws.event.driven.channel.consumer.handler.ProxyEventHandler;
import com.afdor.rws.event.driven.channel.consumer.handler.TagFilter;
import com.afdor.rws.event.driven.commons.ThreadUtils;
import com.afdor.rws.event.driven.config.EventConsumerConfig;
import com.afdor.rws.event.driven.entities.Event;
import lombok.Getter;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.HashSet;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * 消费者抽象实现
 *
 * @author 悭梵
 * @date Created in 2018-08-15 10:40
 */
@Slf4j
public abstract class AbstractEventConsumer<E extends EventHolder> implements EventConsumer, Runnable {
    /**
     * 线程初始化状态
     */
    private volatile boolean init = Boolean.FALSE;
    /**
     * 线程运行状态
     */
    private volatile boolean running = Boolean.FALSE;
    /**
     * 消费者的名称
     */
    @Getter
    @NonNull
    private String consumerName;
    /**
     * 对应事件源
     */
    @Getter
    @NonNull
    private String eventSourceName;
    /**
     * 配置信息
     */
    @Getter
    @NonNull
    private EventConsumerConfig config;
    /**
     * 事件处理
     */
    @Getter
    @NonNull
    private EventHandler eventHandler;
    /**
     * 消费失败重试策略
     */
    @NonNull
    private EventRetryStrategy eventRetryStrategy;
    /**
     * 序列化方式
     */
    @Getter
    @NonNull
    private Serialization serialization;

    @Getter
    private Set<TagFilter> tagFilterList;

    @Override
    public <C extends EventConsumerConfig> C getConfig() {
        return (C) this.config;
    }

    @Override
    public <T extends EventConsumer, C extends EventConsumerConfig> T setConfig(@NonNull C config) {
        checkStatus();
        this.config = config;
        return (T) this;
    }

    @Override
    public <T extends EventConsumer> T setEventSourceName(@NonNull String eventSourceName) {
        checkStatus();
        this.eventSourceName = eventSourceName;
        return (T) this;
    }

    @Override
    public <T extends EventConsumer> T setConsumerName(String consumerName) {
        checkStatus();
        this.consumerName = consumerName;
        return (T) this;
    }

    @Override
    public <T extends EventConsumer> T setEventHandler(@NonNull EventHandler eventHandler) {
        checkStatus();
        this.eventHandler = new ProxyEventHandler(eventHandler);
        return (T) this;
    }

    @Override
    public EventRetryStrategy getEventRetryStrategy() {
        return eventRetryStrategy;
    }

    @Override
    public <T extends EventConsumer> T setEventRetryStrategy(EventRetryStrategy eventRetryStrategy) {
        checkStatus();
        this.eventRetryStrategy = eventRetryStrategy;
        return (T) this;
    }

    @Override
    public <T extends EventConsumer> T setSerialization(@NonNull Serialization serialization) {
        checkStatus();
        this.serialization = serialization;
        return (T) this;
    }

    @Override
    public <T extends EventConsumer> T addTagFilter(TagFilter tagFilter) {
        checkStatus();
        if (this.tagFilterList == null) {
            this.tagFilterList = new HashSet<>(8);
        }
        this.tagFilterList.add(tagFilter);
        return (T) this;
    }

    @Override
    public <T extends EventConsumer> T init() {
        if (getConfig() == null) {
            throw new RuntimeException("消费者配置信息不能为空");
        }
        if (StringUtils.isBlank(getEventSourceName())) {
            throw new RuntimeException("消费者处理事件源不能为空");
        }
        if (getSerialization() == null) {
            throw new RuntimeException("消费者对应事件序列化处理器不能为空");
        }
        if (getEventHandler() == null) {
            throw new RuntimeException("消费者对应事件处理器不能为空");
        }
        if (getEventRetryStrategy() == null) {
            throw new RuntimeException("消费者对应事件重试策略不能为空");
        }
        this.init = Boolean.TRUE;
        return (T) this;
    }

    @Override
    public final void start() {
        if (this.running) {
            return;
        }
        if (!this.init) {
            init();
        }
        doStart();
        this.running = Boolean.TRUE;
        int period = getConfig().getEventHandlerPeriod();
        if (period <= 0) {
            period = 100;
        }
        // scheduleAtFixedRate  在initialDelay时长后第一次执行任务，以后每隔period时长，再次执行任务。注意，period是从任务开始执行算起的。开始执行任务后，定时器每隔period时长检查该任务是否完成，如果完成则再次启动任务，否则等该任务结束后才再次启动任务
        // scheduleWithFixedDelay 在initialDelay时长后第一次执行任务，以后每当任务执行完成后，等待delay时长，再次执行任务。
        // 两个方法均在initialDelay时长后第一次执行任务，以后每隔period时长，再次执行任务，但scheduleAtFixedRate以开始时间计算间隔，而scheduleWithFixedDelay以结束时间计算间隔
        ThreadUtils.getScheduledThreadPool().scheduleWithFixedDelay(this, 0, period, TimeUnit.MILLISECONDS);
    }

    @Override
    public final void stop() {
        doStop();
        this.running = Boolean.FALSE;
        ThreadUtils.getScheduledThreadPool().remove(this);
    }

    @Override
    public void destroy() {
        stop();
        init = Boolean.FALSE;
    }

    @Override
    public final void run() {
        if (this.running && !Thread.currentThread().isInterrupted() && hasEvent()) {
            /* 线程未终止，就执行事件消费。若处理事件返回为true，这继续循环处理事件； 若处理事件返回为false，则退出执行，有计划任务重新启动执行*/
            while (consume()) {
                if (this.running && Thread.currentThread().isInterrupted()) {
                    break;
                }
            }
        }
    }

    /**
     * 启动
     */
    protected void doStart() {

    }

    /**
     * 停止
     */
    protected void doStop() {

    }

    /**
     * 是否存在待消费消息
     *
     * @return
     */
    protected boolean hasEvent() {
        /**
         * 永远有消息
         */
        return true;
    }

    /**
     * 执行标签过滤，返回true标识通过，false标识未通过，当配置多个标签过滤器时，只有当全部都通过时，才能执行事件消费
     *
     * @param event
     * @return
     */
    protected boolean tagFilter(Event event) {
        return this.tagFilterList == null ||
                this.tagFilterList.isEmpty() ||
                Optional.ofNullable(this.tagFilterList)
                        .get()
                        .parallelStream()
                        .allMatch(tagFilter -> tagFilter.doFilter(event));
    }

    /**
     * 获取最新事件
     *
     * @return
     */
    protected E getEvent() {
        return null;
    }

    /**
     * 批量获取最新事件
     *
     * @param batchSize
     * @return
     */
    protected List<E> getEvents(int batchSize) {
        return null;
    }

    /**
     * 事件确认消费成功回调
     *
     * @return
     */
    protected abstract void doSuccess(E event, Object result);

    /**
     * 事件确认消费失败回调
     *
     * @return
     */
    protected abstract void doFailure(E event, String message);


    /**
     * 消费具体消息
     *
     * @return 返回true表示消费到事件，可能还有更多事件待消费；false表示没有消费到事件
     */
    protected boolean consume() {
        // 批量处理
        if (getConfig().getBatchSize() > 1) {
            List<E> eventHolderList = getEvents(getConfig().getBatchSize());
            if (eventHolderList == null || eventHolderList.isEmpty()) {
                return Boolean.FALSE;
            }
            eventHolderList.forEach(eventHolder -> handler(eventHolder));
        } else {
            // 单条处理
            E eventHolder = getEvent();
            if (eventHolder == null || eventHolder.getEvent() == null) {
                return Boolean.FALSE;
            }
            handler(eventHolder);
        }
        return Boolean.TRUE;
    }

    /**
     * 处理单个事件消息
     *
     * @param eventHolder
     */
    private void handler(E eventHolder) {
        if (eventHolder == null) {
            return;
        }
        Event event = eventHolder.getEvent();
        if (event == null) {
            return;
        }
        if (tagFilter(event)) {
            try {
                event.setStatus(Event.Status.EXECUTING);
                Object result = getEventHandler().handler(event);
                doSuccess(eventHolder, result);
                event.setStatus(Event.Status.COMMITTED);
            } catch (Exception e) {
                EventRetryStrategy eventRetryStrategy = getEventRetryStrategy();
                // 返回false表示到达最大重试次数，即认为消息执行失败
                if (eventRetryStrategy != null && eventRetryStrategy.allowRetry(eventHolder)) {
                    event.setStatus(Event.Status.FAILURED);
                    eventRetryStrategy.retry(eventHolder);
                    log.error("事件[" + event.getId() + "]第" + event.getRetryCount() + "次执行失败：" + event + "\n", e);
                } else {
                    event.setStatus(Event.Status.TERMINATED);
                    doFailure(eventHolder, ThreadUtils.getStackMessage(e));
                }
            }
        }
    }

    /**
     * 验证是否启动
     */
    private void checkStatus() {
        if (this.running) {
            throw new RuntimeException("消费者启动后，不能进行修改");
        }
    }

}
